package com.hyt.it.ogt.ks.job;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.hyt.it.ogt.ks.config.ConfigManager;
import com.hyt.it.ogt.ks.enums.CandidateStatusCode;
import com.hyt.it.ogt.ks.feign.KwClient;
import com.hyt.it.ogt.ks.model.entity.CandidateProcess;
import com.hyt.it.ogt.ks.service.ICandidateProcessService;
import com.xxl.job.core.handler.annotation.XxlJob;
import io.seata.common.exception.DataAccessException;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.lang.reflect.Field;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Proxy;
import java.text.SimpleDateFormat;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.util.*;
import java.util.stream.Collectors;

/**
 * @author liql
 * @ClassName CandidateStatusSolveTask.java
 * @Description 考生考中状态处理
 * @createTime 2022年10月09日
 */

@Component
@Slf4j
public class CandidateStatusSolveTask implements InitializingBean {

    @Resource
    private RedisTemplate redisTemplate;
    @Resource
    private ICandidateProcessService iCandidateProcessService;
    @Resource
    private ConfigManager configManager;
    @Resource
    private KwClient kwClient;
    @Value("${ksWebsocketStatusJob.value:}")
    private String ksWebsocketStatusJob;


    private final String WEBSOCKET_NORMAL_KEY="WEBSOCKET_NORMAL_KEY:";
    private final String WEBSOCKET_ERROR_KEY="WEBSOCKET_ERROR_KEY:";
    private final String WEBSOCKET_UNDERLINE="_";
    private final int MAX_UPDATE_NUM=100;

    //处理考生考中状态
    @XxlJob("ksWebsocketStatusJob")
    public void ksWebsocketStatusJob(){
        try {
            //所有考生缓存数据key
            Set<String> normalKeys = redisTemplate.keys(WEBSOCKET_NORMAL_KEY+"*");
            //中断状态考生缓存数据key
            Set<String> errorKeys = redisTemplate.keys(WEBSOCKET_ERROR_KEY+"*");
            //所有考生缓存数据value
            List<String> normalList = redisTemplate.opsForValue().multiGet(normalKeys);
            long milli = LocalDateTime.now().toInstant(ZoneOffset.of("+8")).toEpochMilli();
            List<CandidateProcess> updateList=new ArrayList<>();

            List<CandidateProcess> removeList=new ArrayList<>();
            List<String> delList=new ArrayList<>();
            Map<String,String> saveMap=new HashMap<>();
            Map<String,String> errorKeyMap=new HashMap<>();
            for (String errorKey : errorKeys) {
                errorKeyMap.put(errorKey,errorKey);
            }
            for (String data : normalList) {
                if(StringUtils.isNotEmpty(data)){
                    String examid = data.split(WEBSOCKET_UNDERLINE)[0];
                    String ksid = data.split(WEBSOCKET_UNDERLINE)[1];
                    String timestamp = data.split(WEBSOCKET_UNDERLINE)[2];
                    long timenum = Long.parseLong(timestamp);
                    String ERROR_KEY=WEBSOCKET_ERROR_KEY+examid+WEBSOCKET_UNDERLINE+ksid;
                    if(timenum<(milli-configManager.getWebsocketTimeOut())){
                        if(!errorKeyMap.containsKey(ERROR_KEY)){
                            CandidateProcess process=new CandidateProcess();
                            process.setExamId(examid);
                            process.setCandidateId(ksid);
                            updateList.add(process);
                            saveMap.put(ERROR_KEY,examid+WEBSOCKET_UNDERLINE+ksid+WEBSOCKET_UNDERLINE+LocalDateTime.now().toInstant(ZoneOffset.of("+8")).toEpochMilli());
                        }
                    }else if(errorKeyMap.containsKey(ERROR_KEY)){
                        delList.add(ERROR_KEY);
                        CandidateProcess process=new CandidateProcess();
                        process.setExamId(examid);
                        process.setCandidateId(ksid);
                        removeList.add(process);
                    }
                }
            }
            if(!saveMap.isEmpty()){
                redisTemplate.opsForValue().multiSet(saveMap);
                batchSetOrExpire(saveMap,3600);
            }
            if(CollectionUtils.isNotEmpty(delList)){
                redisTemplate.delete(delList);
            }
            //考中变为中断
            if(CollectionUtils.isNotEmpty(updateList)){
                if(updateList.size()>MAX_UPDATE_NUM){
                    //避免因过多中断考生数据更新时间过长 造成的锁表
                    log.error("中断考生数量过多，请联系系统管理人员进行查验！！！中断考生数量："+updateList.size());
                }else{
                    LambdaQueryWrapper<CandidateProcess> queryWrapper = Wrappers.<CandidateProcess>lambdaQuery().in(CandidateProcess::getCandidateId, updateList.stream().map(CandidateProcess::getCandidateId).collect(Collectors.toList()))
                            .in(CandidateProcess::getExamId, updateList.stream().map(CandidateProcess::getExamId).collect(Collectors.toList()))
                            .and(wrapper->wrapper.eq(CandidateProcess::getExamStatus, CandidateStatusCode.TESTING.getCode()).or(j->j.eq(CandidateProcess::getExamStatus, CandidateStatusCode.INTERRUPT.getCode())));
                    List<CandidateProcess> list = iCandidateProcessService.list(queryWrapper);
                    if(CollectionUtils.isNotEmpty(list)){
                        for (CandidateProcess process : list) {
                            for (CandidateProcess upprocess : updateList) {
                                if(process.getExamId().equals(upprocess.getExamId())&&process.getCandidateId().equals(upprocess.getCandidateId())){
                                    kwClient.updateCandidateOnlineExamState(process.getExamId(), process.getCandidateId(), true, CandidateStatusCode.INTERRUPT.getCode());
                                    break;
                                }
                            }
                        }
                    }
                    LambdaUpdateWrapper<CandidateProcess> update = Wrappers.<CandidateProcess>lambdaUpdate()
                            .set(CandidateProcess::getExamStatus, CandidateStatusCode.INTERRUPT.getCode())
                            .eq(CandidateProcess::getExamStatus, CandidateStatusCode.TESTING.getCode())
                            .in(CandidateProcess::getCandidateId, updateList.stream().map(CandidateProcess::getCandidateId).collect(Collectors.toList())).in(CandidateProcess::getExamId, updateList.stream().map(CandidateProcess::getExamId).collect(Collectors.toList()));
                    iCandidateProcessService.update(update);
                }
            }
            //中断改为考中
            if(CollectionUtils.isNotEmpty(removeList)){
                LambdaQueryWrapper<CandidateProcess> queryWrapper = Wrappers.<CandidateProcess>lambdaQuery().in(CandidateProcess::getCandidateId, removeList.stream().map(CandidateProcess::getCandidateId).collect(Collectors.toList()))
                        .in(CandidateProcess::getExamId, removeList.stream().map(CandidateProcess::getExamId).collect(Collectors.toList()))
                        .eq(CandidateProcess::getExamStatus, CandidateStatusCode.INTERRUPT.getCode());
                List<CandidateProcess> list = iCandidateProcessService.list(queryWrapper);
                if(CollectionUtils.isNotEmpty(list)){
                    for (CandidateProcess process : list) {
                        for (CandidateProcess upprocess : removeList) {
                            if(process.getExamId().equals(upprocess.getExamId())&&process.getCandidateId().equals(upprocess.getCandidateId())){
                                kwClient.updateCandidateOnlineExamState(process.getExamId(), process.getCandidateId(), true, CandidateStatusCode.TESTING.getCode());
                                break;
                            }
                        }
                    }
                }
                LambdaUpdateWrapper<CandidateProcess> update = Wrappers.<CandidateProcess>lambdaUpdate()
                        .set(CandidateProcess::getExamStatus, CandidateStatusCode.TESTING.getCode())
                        .eq(CandidateProcess::getExamStatus, CandidateStatusCode.INTERRUPT.getCode())
                        .in(CandidateProcess::getCandidateId, removeList.stream().map(CandidateProcess::getExamId).collect(Collectors.toList())).in(CandidateProcess::getExamId, removeList.stream().map(CandidateProcess::getExamId).collect(Collectors.toList()));
                iCandidateProcessService.update(update);
            }
        }catch (Exception e){
            log.error("ksWebsocketStatusJob定时任务执行失败！"+e.getMessage());
        }
    }
    /**
     * 批量添加 并且设置失效时间
     * @param map
     * @param seconds
     */
    public void batchSetOrExpire(Map<String, String> map, long seconds) {
        RedisSerializer<String> serializer = redisTemplate.getStringSerializer();
        redisTemplate.executePipelined(new RedisCallback<String>() {
            @Override
            public String doInRedis(RedisConnection connection) throws DataAccessException {
                map.forEach((key, value) -> {
                    connection.expire(serializer.serialize(key),  seconds);
                });
                return null;
            }
        }, serializer);
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        if(!com.hyt.core.util.StringUtils.isEmpty(ksWebsocketStatusJob)) {
            XxlJob xxlJob = this.getClass().getDeclaredMethod("ksWebsocketStatusJob").getAnnotation(XxlJob.class);
            InvocationHandler invocationHandler = Proxy.getInvocationHandler(xxlJob);
            Field declaredField = invocationHandler.getClass().getDeclaredField("memberValues");
            declaredField.setAccessible(true);
            Map memberValues = (Map) declaredField.get(invocationHandler);
            memberValues.put("value", xxlJob.value() + ksWebsocketStatusJob);
        }
    }

    public static void main(String[] args) {
        long milliSecond = 1666576630678L;
        Date date = new Date();
        date.setTime(milliSecond);
        System.out.println(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(date));
    }
}
